#pragma once
#include <mqcomm/Helper.hpp>
#include <mqcomm/Log.hpp>
#include <mqcomm/Message.pb.h>
#include <mqcomm/protocol.pb.h>
#include <muduo/net/TcpClient.h>
#include <muduo/net/EventLoopThread.h>
#include <muduo/protobuf/codec.h>
#include <muduo/protobuf/dispatcher.h>
#include <muduo/base/Logging.h>
#include <muduo/base/Mutex.h>
#include <muduo/base/CountDownLatch.h>
#include "workLoop.hpp"
#include "channel.hpp"
#include <iostream>
#include <string>
#include <memory>
using namespace std;

namespace rbmq {
    class connection {
    public :
        typedef shared_ptr<connection> ptr;
        typedef shared_ptr<google::protobuf::Message> MessagePtr;
        // typedef shared_ptr<mqNs::basicCommonResponse> basicCommonResponsePtr;
        // typedef shared_ptr<mqNs::basicConsumerRequest> basicConsumerRequestPtr;
        connection(const string& serverip, int serverport, const AsyncWorker::ptr& worker) 
            :_latch(1)
            ,_async_worker(worker)
            ,_client(_async_worker->_loopthread.startLoop(), muduo::net::InetAddress(serverip, serverport), "connection")
            ,_dispatcher(std::bind(&connection::OnUnkownMessage, this, placeholders::_1, placeholders::_2, placeholders::_3))
            ,_codec(make_shared<ProtobufCodec>(std::bind(&ProtobufDispatcher::onProtobufMessage, &_dispatcher,\
             placeholders::_1, placeholders::_2, placeholders::_3)))
            ,_channel_manager(make_shared<channelManager>())
        {
            _dispatcher.registerMessageCallback<mqNs::basicCommonResponse>(std::bind(&connection::onBasicResponse, this,\
             placeholders::_1, placeholders::_2, placeholders::_3));
            _dispatcher.registerMessageCallback<mqNs::basicConsumerResponse>(std::bind(&connection::onBasicConsumerResponse, this,\
             placeholders::_1, placeholders::_2, placeholders::_3));

            _client.setConnectionCallback(std::bind(&connection::OnConnection, this, placeholders::_1));
            _client.setMessageCallback(std::bind(&ProtobufCodec::onMessage, _codec.get(), placeholders::_1, placeholders::_2, placeholders::_3));

        }

        void connect() {
            _client.connect(); // 开始连接服务器
            _latch.wait(); // 阻塞等待，直到连接到来
        }

        // 打开一个信道
        channel::ptr openChannel() {
            // 在创建信道后，还应该给服务端发送一个创建信道的请求
            // 使用_channel_manager创建出一个信道
            channel::ptr chn = _channel_manager->createChannel(_conn, _codec);

            if (chn.get() == nullptr) {
                LOG(Debug) << "打开信道失败!" << endl;
                return channel::ptr();
            }
            // 通过返回的信道对象给服务端发送一个创建信道的请求
            bool res = chn->openChannel();
            if (!res) {
                LOG(Debug) << "服务端创建信道失败!" << endl;
            }
            LOG(Info) << "创建信道成功, 信道id: " << chn->getChannelId() << endl;
            return chn;
        }

        // 关闭一个信道
        void closeChannel(const channel::ptr& chn) {
            chn->closeChannel();
            _channel_manager->removeChannel(chn->getChannelId());
        }

    private :
        bool send(const google::protobuf::Message* message) {
            if (_conn->connected()) {
                // 发送
                _codec->send(_conn, *message);
                return true;
            }
            return false;
        }



        void onBasicResponse(const muduo::net::TcpConnectionPtr& conn, const basicCommonResponsePtr resp, muduo::Timestamp) {
            // 1、找到信道
            channel::ptr chn = _channel_manager->getChannel(resp->cid());
            if (chn.get() == nullptr) {
                LOG(Debug) << "处理统一响应的时候, 没有找到信道id为: " << resp->cid() << " 的信道!" << endl;
                return;
            }
            // 2、将得到的响应对象，通过信道，添加到信道的基础响应哈希表中
            chn->putbasicResponse(resp);
        }

        void onBasicConsumerResponse(const muduo::net::TcpConnectionPtr& conn, const basicConsumerResponsePtr resp, muduo::Timestamp) {
            // 1、找到信道
            channel::ptr chn = _channel_manager->getChannel(resp->cid());
            if (chn.get() == nullptr) {
                LOG(Debug) << "没有找到对应的信道!" << endl;
                return;
            }
            // 2、封装一个异步任务，抛入到线程池中  
            // 这里千万不能使用引用的方式，捕获智能指针，因为使用引用捕获并不会引起引用计数的增加，
            // 所以当前作用域的chn在函数结束后就销毁了(引用计数变为0)，
            // 那么在lambda表达式内部的chn就可能是一个野指针(因为是两个线程并发在跑，哪一个先跑完不能确定)
            _async_worker->_threadpool.push([chn, resp](){
                // 封装的任务要做的是：让找到的线程执行消费任务
                chn->consumer(resp);
            });
        }

        void OnUnkownMessage(const muduo::net::TcpConnectionPtr& conn, const MessagePtr message, muduo::Timestamp) {
            conn->shutdown();
        }
        void OnConnection(const muduo::net::TcpConnectionPtr& conn) {
            if (conn->connected()) {
                _conn = conn;
                _latch.countDown(); // 唤醒connect里面的阻塞
            } else {
            }
        }

    private :
        // muduo::net::EventLoopThread _loopthread;
        AsyncWorker::ptr _async_worker;
        muduo::net::TcpClient _client;
        muduo::CountDownLatch _latch;
        muduo::net::TcpConnectionPtr _conn;
        ProtobufCodecPtr _codec;
        ProtobufDispatcher _dispatcher;
        channelManager::ptr _channel_manager; // 信道管理句柄集合
    };
   
};